package com.hzgj.bcl.redis;

import com.google.common.base.Charsets;
import com.hzgj.bcl.util.lang.StrKit;
import lombok.Setter;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedisPipeline;
import redis.clients.jedis.Tuple;
import redis.clients.jedis.exceptions.JedisException;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.*;

public final class RedisClient {
    private static final Logger logger = LoggerFactory.getLogger(RedisClient.class);
    private ShardedXJedisPool pool;
    @Setter
    private int minPoolSize = 1;
    @Setter
    private int maxPoolSize = 100;
    @Setter
    private int connectTimeout = 2000;
    @Setter
    private int readTimeout = 2000;
    @Setter
    List<RedisServerInfo> servers;


    @PostConstruct
    public void init() {
        pool = getPool();
    }

    private synchronized ShardedXJedisPool getPool() {
        if (pool != null) {
            return pool;
        }
        ArrayList<JedisShardInfo> shards = new ArrayList<>();
        Objects.requireNonNull(servers);
        servers.forEach(redisServerInfo -> {
            JedisShardInfo shard = new JedisShardInfo(redisServerInfo.host, redisServerInfo.port);
            if (StrKit.notBlank(redisServerInfo.password)) {
                shard.setPassword(redisServerInfo.password);
            }
            if (connectTimeout > 0) {
                shard.setConnectionTimeout(connectTimeout);
            }
            if (readTimeout > 0) {
                shard.setSoTimeout(readTimeout);
            }

            shards.add(shard);
        });

        GenericObjectPoolConfig cfg = new GenericObjectPoolConfig();
        cfg.setMaxTotal(maxPoolSize);
        cfg.setMinIdle(minPoolSize);
        pool = new ShardedXJedisPool(cfg, shards);
        return pool;
    }

    public void shutdown() {
        if (pool != null) {
            pool.close();
        }
    }


    /**
     * 获取值
     *
     * @param key
     * @return
     */
    public String get(String key) {
        return execute(key, jedis -> jedis.get(key));
    }


    /**
     * 设置值并返回之前的值
     *
     * @param key
     * @param value
     * @return
     */
    public String getSet(String key, String value) {
        return execute(key, jedis -> jedis.getSet(key, value));
    }

    /**
     * 添加缓存, 仅当键不存在时才会插入
     *
     * @param key   键
     * @param value 值
     * @return
     */
    public boolean add(String key, String value) {
        return execute(key, jedis -> {
            long result = jedis.setnx(key, value);
            return result == 1;
        });
    }

    /**
     * 设置缓存
     *
     * @param key   键
     * @param value 值
     * @return
     */
    public String set(String key, String value) {
        return execute(key, jedis -> jedis.set(key, value));
    }

    /**
     * 设置缓存
     *
     * @param key    键
     * @param value  值
     * @param expiry 过期时间
     * @return
     */
    public String set(String key, String value, Duration expiry) {
        return execute(key, jedis -> jedis.setex(key, (int) expiry.getSeconds(), value));
    }

    /**
     * 是否存在缓存
     *
     * @param key 缓存键
     */
    public boolean exist(String key) {
        return execute(key, jedis -> jedis.exists(key));
    }

    /**
     * 删除缓存
     *
     * @param key 缓存键
     */
    public void delete(String key) {
        execute(key, jedis -> jedis.del(key));
    }

    /**
     * 增加缓存值 1
     *
     * @param key 缓存键
     */
    public long increase(String key) {
        return execute(key, jedis -> jedis.incr(key));
    }

    /**
     * 增加缓存值 value
     *
     * @param key   缓存键
     * @param value 增加的值
     */
    public long increase(String key, long value) {
        return execute(key, jedis -> jedis.incrBy(key, value));
    }

    public long zadd(String key, double score, String value) {
        return execute(key, jedis -> jedis.zadd(key, score, value));
    }

    public long zadd(String key, Map<String, Double> memberAndScore) {
        return execute(key, jedis -> jedis.zadd(key, memberAndScore));
    }

    public Set<String> zrange(String key, long start, long end) {
        return execute(key, jedis -> jedis.zrange(key, start, end));
    }

    public Set<String> zrevrange(String key, long start, long stop) {
        return execute(key, jedis -> jedis.zrevrange(key, start, stop));
    }

    public long zrem(String key, String... members) {
        return execute(key, jedis -> jedis.zrem(key, members));
    }

    public long zremrangeByRank(String key, long start, long end) {
        return execute(key, jedis -> jedis.zremrangeByRank(key, start, end));
    }

    public long zremrangeByScore(String key, double start, double end) {
        return execute(key, jedis -> jedis.zremrangeByScore(key, start, end));
    }

    public long zcard(String key) {
        return execute(key, jedis -> jedis.zcard(key));
    }

    public long sadd(String key, String... members) {
        return execute(key, jedis -> jedis.sadd(key, members));
    }

    public Set<String> smembers(String key) {
        return execute(key, jedis -> jedis.smembers(key));
    }

    public Long srem(String key, String... members) {
        return execute(key, jedis -> jedis.srem(key, members));
    }

    public Set<Tuple> zrangeByScoreWithScores(String key, double min, double max) {
        return execute(key, jedis -> jedis.zrangeByScoreWithScores(key, min, max));
    }

    public Set<String> zrevrangebyscore(String key, double max, double min, int offset, int count) {
        return execute(key, jedis -> jedis.zrevrangeByScore(key, max, min, offset, count));
    }

    public Long zcount(String key, double min, double max) {
        return execute(key, jedis -> jedis.zcount(key, min, max));
    }

    public Long hset(byte[] key, byte[] field, byte[] value) {
        return execute(key, jedis -> jedis.hset(key, field, value));
    }

    public boolean hexists(byte[] key, byte[] field) {
        return execute(key, jedis -> jedis.hexists(key, field));
    }

    public Long hdel(byte[] key, byte[] field) {
        return execute(key, jedis -> jedis.hdel(key, field));
    }

    public byte[] hget(byte[] key, byte[] field) {
        return execute(key, jedis -> jedis.hget(key, field));
    }

    public Long hlen(byte[] key) {
        return execute(key, jedis -> jedis.hlen(key));
    }

    public Set<byte[]> hkeys(byte[] key) {
        return execute(key, jedis -> jedis.hkeys(key));
    }

    public <T> List<String> subscribe(RedisSubScribeListener redisSubScribeListener, String channel) {
        return execute(channel, jedis -> jedis.subscribe(redisSubScribeListener, channel));
    }

    public Long publish(String channel, String message) {
        return execute(channel, jedis -> jedis.publish(channel, message));
    }

    public Object eval(String script, List<String> keys, List<String> args) {
        return execute(() -> MD5Hash.Instance.hash(script.getBytes(Charsets.UTF_8)), jedis -> jedis.script(script, keys, args));
    }

    /**
     * 给一个key设置过期时间,以秒来计算，比如10秒后key过期<br>
     *
     * @param seconds
     * @return boolean
     */
    public boolean setKeyExpireTime(String key, int seconds) {
        return execute(key, jedis -> {
            if (key != null && !key.trim().equals("") && seconds > 0) {
                jedis.expire(key, seconds);
                return Boolean.TRUE;
            }
            return Boolean.FALSE;
        });
    }

    public void pipeline(PipelineCallback callback) {

        ShardedXJedis jedis = pool.getResource();
        try {
            ShardedJedisPipeline pipeline = jedis.pipelined();
            callback.call(pipeline);
            pipeline.sync();
        } catch (JedisException e) {
            logger.error("pipeline failed, error: {}", e);
            throw e;
        } finally {
            jedis.close();
        }
    }

    public interface PipelineCallback {
        void call(ShardedJedisPipeline pipeline);
    }


    protected <Result> Result execute(String key, JedisOperation<Result> op) {
        return execute(key.getBytes(Charsets.UTF_8), op);
    }

    protected <Result> Result execute(KeyOperation key, JedisOperation<Result> op) {
        ShardedXJedis jedis = pool.getResource();
        try {
            return op.call(jedis);
        } catch (JedisException e) {
            logger.error("operation failed, server: {}, error: {}", jedis.getShardInfo(key.getKey()), e);
            throw e;
        } finally {
            jedis.close();
        }
    }

    protected <Result> Result execute(byte[] key, JedisOperation<Result> op) {
        ShardedXJedis jedis = pool.getResource();
        try {
            return op.call(jedis);
        } catch (JedisException e) {
            logger.error("operation failed, server: {}, error: {}", jedis.getShardInfo(key), e);
            throw e;
        } finally {
            jedis.close();
        }
    }
}
